package com.danbay.framework.kafka.config;

import com.danbay.framework.kafka.AccessPackageSerializer;
import com.danbay.framework.kafka.KafkaClient;
import com.danbay.framework.kafka.coap.CoAPJsonDataDecoder;
import com.danbay.framework.kafka.coap.CoAPJsonDataEncoder;
import com.danbay.framework.redis.config.RedisConfigurator;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;

/**
 * @author SJG
 *         2017/11/11.
 */
@AutoConfigureAfter(KafkaConfiguration.class)
@ImportAutoConfiguration(RedisConfigurator.class)
@EnableConfigurationProperties(KafkaProperties.class)
public class KafkaProducerConfig {

    final KafkaProperties kafkaProperties;

    @Autowired
    public KafkaProducerConfig(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
    }

    public DefaultKafkaProducerFactory producerFactory() {
        return new DefaultKafkaProducerFactory(kafkaProperties.buildProducerProperties(), new StringSerializer(), new AccessPackageSerializer());
    }

    @Bean
    public KafkaClient getKafkaClient() {
        return new KafkaClient<>(producerFactory());
    }

    @ConditionalOnMissingBean
    @Bean
    public AutoKafkaListener getAutoKafkaListener() {
        return new AutoKafkaListener(new CoAPJsonDataDecoder(), new CoAPJsonDataEncoder());
    }
}
